package com.ry.flink.jobx;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;

import java.time.Instant;

/**
 *  8. 关于 changelogStream 方法应用
 *        Upsert 流和Retract 流对应的操作
 * */
public class Job8 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tablEnv = StreamTableEnvironment.create(env);

        DataStreamSource<Row> dataStreamSource = env.fromElements(
                Row.ofKind(RowKind.INSERT, "貂蝉", 85),
                Row.ofKind(RowKind.INSERT, "孙尚香", 87),
                Row.ofKind(RowKind.UPDATE_BEFORE, "貂蝉", 85),
                Row.ofKind(RowKind.UPDATE_AFTER, "貂蝉",98));
        Table table = tablEnv.fromChangelogStream(dataStreamSource);
        table.execute().print();

        tablEnv.createTemporaryView("dreamTable",dataStreamSource);

        tablEnv.executeSql("SELECT f0 AS name,SUM(f1) score FROM dreamTable GROUP BY f0")
                .print();
        /**
         * Retract 流：这种方式对于觉大部分的场景都适用，因为它能够接收所有类型的更改。
         * +----+--------------------------------+-------------+
         * | op |                             f0 |          f1 |
         * +----+--------------------------------+-------------+
         * | +I |                           貂蝉 |          85 |
         * | +I |                         孙尚香 |          87 |
         * | -U |                           貂蝉 |          85 |
         * | +I |                           貂蝉 |          98 |
         * +----+--------------------------------+-------------+
         *
         *
         *
         * */

        /**
         * Upsert 流:
         *   1. 需要定义一个主键
         *   2. 指定模式为upsert更改日志模式
         * */

        DataStreamSource<Row> dataStreamSourceUpsert = env.fromElements(
                Row.ofKind(RowKind.INSERT, "貂蝉", 85),
                Row.ofKind(RowKind.INSERT, "孙尚香", 87),
                Row.ofKind(RowKind.UPDATE_AFTER, "貂蝉",98));

        Table table2 = tablEnv.fromChangelogStream(
                dataStreamSourceUpsert,
                Schema.newBuilder().primaryKey("f0").build(),
                ChangelogMode.upsert());

        tablEnv.createTemporaryView("dataStreamSourceUpsert",dataStreamSourceUpsert);

        tablEnv.executeSql("SELECT f0 AS name,SUM(f1) score FROM dataStreamSourceUpsert GROUP BY f0");
//                .print();

        /**
         * 基于Upsert的结果：可以减少结果消息的数量
         * +----+--------------------------------+-------------+
         * | +I |                           貂蝉 |          85 |
         * | +I |                         孙尚香 |          87 |
         * | -U |                           貂蝉 |          85 |
         * | +U |                           貂蝉 |          98 |
         * +----+--------------------------------+-------------+
         *
         * */
        table2.execute().print();

    }
}
